package com.booter.service;

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.util.CollectionUtils;

import java.io.Closeable;
import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * @author xueqiu
 * @date 2023/3/28 18:28
 * HBase2.1.2之后的版本，无需创建HBase线程池，HBase2.1.2提供的代码已经封装好，只需创建调用即可
 * table在用完之后直接调用close方法即可，程序关闭时调用conn的close方法
 */
@Slf4j
public class HbaseTemplate implements Closeable {

    private Connection connection;

    private Map<String, Map<String, Class<?>>> columns = new ConcurrentHashMap<>();

    /**
     * hbase.zookeeper.port(hbase.zookeeper.property.clientPort)=2181
     * hbase.zookeeper.quorum=10.10.10.10
     * hbase.hconnection.threads.max=128 最大线程池(相当于连接池)
     * hbase.client.scanner.max.result.size HBase-1.2默认值为2097152，即2M。Scan.next()的一次RPC请求fetch的数据量大小，HBase-1.2使用这个参数控制RPC次数和流量。
     * hbase.server.scanner.max.result.size HBase-1.2默认值为104857600，即10M。该参数表示当Scan.next()发起RPC后，服务端返回给客户端的最大字节数，防止Server OOM。
     * hbase.client.scanner.timeout.period 一次Scan中一次RPC请求的超时时间（一次Scan可能有多次RPC请求）
     * hbase.client.operation.timeout 客户端发起一次数据操作直至得到响应之间总的超时时间，数据操作类型包括get、append、increment、delete、put等。
     * hbase.rpc.timeout 一次RPC请求的超时时间,超过该值客户端就会主动关闭socket。服务端会抛出: java.io.IOException: Connection reset by peer
     * hbase.zookeeper.znode
     * zookeeper.znode.parent 默认: /hbase ZooKeeper中的HBase的根ZNode。所有的HBase的ZooKeeper会用这个目录配置相对路径。默认情况下，所有的HBase的ZooKeeper文件路径是用相对路径，所以他们会都去这个目录下面。
     * hbase.client.retries
     * hbase.client.retries.number
     * zookeeper.recovery.retry
     * hbase.client.keyvalue.maxsize=83886080 KeyValue实例的最大size.设置存储文件中的单个entry的大小上界。KeyValue是不能分割的可以避免因为数据过大导致region不可分割。明智的做法是把它设为可以被最大region size整除的数。如果设置为0或者更小，就会禁用这个检查。默认10MB
     * hbase.rootdir
     * <p>
     * 在scan过程中RegionServer端偶尔抛出leaseException,HBase内部在一次完整的scan操作中引入了租约机制,完整的scan通常会被拆分为多个RPC请求，
     * RegionServer接收到第一次RPC请求之后，会为该scan操作生成一个全局唯一的id称为scanId。除此之外，RegionServer还会进行大量的准备工作，构建整个scan体系(构造需要用到的所有对象),
     * 后续的RPC请求只需要携带相同的scanId作为标示就可以直接利用这些已经构建好的资源进行检索。在整个scan过程中，客户端其实都占用着服务器端的资源，此时如果此客户端意外宕机，是否就意味着这些资源永远都不能得到释放呢？
     * 租约机制就是为了解决这个问题。RegionServer接收到第一次RPC之后，除了生成全局唯一的scanId之外还会生成一个携带有超时时间的lease，超时时间可以通过参数hbase.regionserver.lease.period配置，
     * 一旦在超时时间内后续RPC请求没有到来（比如客户端处理太慢），RegionServer就认为客户端出现异常，此时会将该lease销毁并将整个scan所持有的资源全部释放，客户端在处理完成之后再发后续的RPC过来，
     * 检查到对应的lease已经不存在，就会抛出LeaseException异常。
     *
     * @param config
     */
    public HbaseTemplate(org.apache.hadoop.conf.Configuration config) {
        try {
            connection = ConnectionFactory.createConnection(config);
        } catch (Exception ex) {
            throw new RuntimeException(ex);
        }
    }


    public void addColumns(String table, Map<String, Class<?>> clsMap) {
        this.columns.put(table, clsMap);
    }

    public void createNamespace(String namespace) {
        try (Admin admin = connection.getAdmin()) {
            NamespaceDescriptor desc = NamespaceDescriptor.create(namespace).build();
            admin.createNamespace(desc);
            log.info("namespace {} is create success!", namespace);
        } catch (IOException e) {
            log.error("", e);
        }
    }

    public void createTable(String tableName, List<String> columnFamily) {
        try (Admin admin = connection.getAdmin()) {
            List<ColumnFamilyDescriptor> cfDescriptor = columnFamily.stream().map(e -> ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(e)).build()).collect(Collectors.toList());
            TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName)).setColumnFamilies(cfDescriptor).build();
            if (admin.tableExists(TableName.valueOf(tableName))) {
                log.warn("table {} is exists!", tableName);
            } else {
                admin.createTable(tableDescriptor);
                log.info("table {} is create success!", tableName);
            }
        } catch (IOException e) {
            log.error("", e);
        }
    }

    public void put(String table, Object rowKey, String family, Map<String, Object> data) {
        long ts = System.currentTimeMillis();
        try (Table tb = connection.getTable(TableName.valueOf(table))) {
            //HTable实例并不是线程安全的,并发场景要注意隔离
            Put put = new Put(HbaseSerialize.serialize(rowKey).getValue());
            //防止bean初始化时未能将元信息初始化到columns中，在插入数据时重新写入一次
            boolean flag = !columns.containsKey(table);
            Map<String, Class<?>> clsMap = new HashMap<>();
            data.forEach((k, v) -> {
                if (v != null) {//过滤空值
                    ValueMeta mata = HbaseSerialize.serialize(v);
                    if (flag) {
                        clsMap.put(k, mata.getCls());
                    }
                    put.addColumn(Bytes.toBytes(family), Bytes.toBytes(k), ts, mata.getValue());
                }
            });
            if (flag) {
                columns.put(table, clsMap);
            }
            tb.put(put);
        } catch (Exception e) {
            log.error("", e);
        }
    }

    public Map<String, Object> findOne(String table, String rowKey, String family) {
        try (Table tb = connection.getTable(TableName.valueOf(table))) {
            Get get = buildGet(Bytes.toBytes(rowKey), Bytes.toBytes(family), null);
            Result result = tb.get(get);
            return toObjectMetaMap(result.rawCells(), table);
        } catch (Exception e) {
            log.error("", e);
            return Collections.emptyMap();
        }
    }


    private Get buildGet(byte[] rowKey, byte[] family, List<String> qualifiers) {
//        Filter filter = new RowFilter(CompareOperator.LESS_OR_EQUAL,new BinaryComparator(Bytes.toBytes("q")));
        Get get = new Get(rowKey);
        get.addFamily(family);
//        get.readVersions(1);//maxVersion
//        get.setPriority(HConstants.HIGH_QOS);
        if (!CollectionUtils.isEmpty(qualifiers)) {
            qualifiers.forEach(s -> get.addColumn(family, Bytes.toBytes(s)));
        }
        return get;
    }

    private Get buildGet(byte[] rowKey, byte[] family, List<String> qualifiers, int version) throws IOException {
        Get get = new Get(rowKey);
        get.addFamily(family);
//        get.setPriority(HConstants.HIGH_QOS);
        get.readVersions(version);//maxVersion
        if (!CollectionUtils.isEmpty(qualifiers)) {
            qualifiers.forEach(s -> get.addColumn(family, Bytes.toBytes(s)));
        }
        return get;
    }


    private Map<String, Object> toObjectMetaMap(Cell[] cells, String table) {
        Map<String, Object> map = new HashMap<>();
        //bean的元数据未初始化会导致查询无数据
        Map<String, Class<?>> clsMap = columns.get(table);
        if (CollectionUtils.isEmpty(clsMap)) {
            return Collections.emptyMap();
        }
        for (Cell c : cells) {
            String key = Bytes.toString(CellUtil.cloneQualifier(c));
            Class<?> cls = clsMap.get(key);
            if (cls != null) {
                map.put(key, HbaseSerialize.deserialize(CellUtil.cloneValue(c), cls));
            }
        }
        return map;
    }


    @Override
    public void close() throws IOException {
        if (connection != null && !connection.isClosed()) {
            connection.close();
        }
    }
}
